#include "XThread.h"
#include "XTask.h"
#include <event2/event.h>
#include <iostream>
#ifdef _WIN32
#define read(fd,buf,len) recv(fd,buf,len,0)
#define write(fd,buf,len) send(fd,buf,len,0)
#else
#include <unistd.h>
#endif
#include <thread>
#include <chrono>
XThread::XThread()
{
}

XThread::~XThread()
{
    if (m_base)
    {
        event_base_free(m_base);
    }
}

bool XThread::setup()
{
    int ret =0;
    // 创建管道或套接字
    evutil_socket_t fds[2];
#ifndef _WIN32
    ret = pipe(fds);
    if(ret !=0){
        std::cerr<<"pipe"<<std::endl;
        return false;
    }
#else
    ret = evutil_socketpair(AF_INET,SOCK_STREAM,0,fds);
    if(ret < 0){
        std::cerr<<"evutil_socketpair"<<std::endl;
        return false;
    }
    evutil_make_socket_nonblocking(fds[0]);
    evutil_make_socket_nonblocking(fds[1]);
#endif
    notify_send_fd_ = fds[1];
    //创建 libevent上下文 (无锁)
    event_config * ev_confg = event_config_new();
    event_config_set_flag(ev_confg,EVENT_BASE_FLAG_NOLOCK);
    m_base = event_base_new_with_config(ev_confg);
    event_config_free(ev_confg);
    if(!m_base){
        std::cerr<<"event_base_new_with_config"<<std::endl;
        return false;
    }
    event * ev= event_new(m_base,fds[0],EV_READ|EV_PERSIST,XThread::notify_cb,this);
    ret=event_add(ev,nullptr);
    if(ret < 0){
        std::cerr<<"event_add"<<std::endl;
        return false;
    }
    std::cout << "XThread::setup" << std::endl;
    return true;
}

bool XThread::start()
{
    if(!setup()){
        std::cerr<<"XThread::setup err"<<std::endl;
        return false;
    }
    m_thread = std::thread(&XThread::main,this);
    m_thread.detach();
    return true;
}

bool XThread::activate()
{
    char c = 'a';
    int ret = write(notify_send_fd_,&c,1);
    if(ret <= 0){
        std::cerr<<"XThread::write"<<std::endl;
        return false;
    }
    std::cout<<"XThread::activate"<<std::endl;
    return true;
}

void XThread::addTask(XTask* task)
{
    if(!task){
        return;
    }
    task->setBase(this->m_base);
    m_mutex.lock();
    m_taskList.push_back(task);
    m_mutex.unlock();
    std::cout<<"XThread::addTask"<<std::endl;
}

void XThread::main()
{
    std::cout<<std::this_thread::get_id()<<" XThread::main"<<std::endl;
    if(!m_base)
        std::cout<<"XThread id="<<std::this_thread::get_id()<<" XThread::main err"<<std::endl;
    // 循环执行任务，不阻塞
    while (!is_exit_)
    {
        event_base_loop(m_base,EVLOOP_NONBLOCK|EVLOOP_ONCE);
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
    event_base_free(m_base);
    m_base=nullptr;
}

void XThread::notify_cb(evutil_socket_t fd,short flag,void *xthread)
{
    XThread* pXThread = (XThread*)xthread;
    char c = 0;
    int ret=read(fd,&c,1);
    if(ret<=0){
        return;
    }
    std::cout<<"notify_cb"<<std::endl;
    std::unique_lock lock(pXThread->m_mutex);
    if(pXThread->m_taskList.empty()){
        return;
    }
    // 获取任务并初始化任务
    auto task=pXThread->m_taskList.front();
    pXThread->m_taskList.pop_front();
    lock.unlock();
    task->Init();
}
